package com.xiaochao.bigdata.spark.accumulator

import com.xiaochao.bigdata.constant.Constants
import com.xiaochao.bigdata.util.StringUtils
import org.apache.spark.util.AccumulatorV2
/**
  * Scala中，自定义Accumulator
  * 使用object，直接定义一个伴生对象即可
  * 需要实现AccumulatorParam接口，并使用[]语法，定义输入输出的数据格式
  */
class SessionAggrStatAccumulator extends AccumulatorV2[String,String] {


  private var sessionAggeStatCount: String = Constants.SESSION_COUNT + "=0|" + Constants.TIME_PERIOD_1s_3s + "=0|" + Constants.TIME_PERIOD_4s_6s + "=0|" + Constants.TIME_PERIOD_7s_9s + "=0|" + Constants.TIME_PERIOD_10s_30s + "=0|" + Constants.TIME_PERIOD_30s_60s + "=0|" + Constants.TIME_PERIOD_1m_3m + "=0|" + Constants.TIME_PERIOD_3m_10m + "=0|" + Constants.TIME_PERIOD_10m_30m + "=0|" + Constants.TIME_PERIOD_30m + "=0|" + Constants.STEP_PERIOD_1_3 + "=0|" + Constants.STEP_PERIOD_4_6 + "=0|" + Constants.STEP_PERIOD_7_9 + "=0|" + Constants.STEP_PERIOD_10_30 + "=0|" + Constants.STEP_PERIOD_30_60 + "=0|" + Constants.STEP_PERIOD_60 + "=0"


  override def isZero: Boolean = {
    true
  }



  override def copy(): AccumulatorV2[String, String] = {

    val newMyAcc = new SessionAggrStatAccumulator()
    newMyAcc.sessionAggeStatCount = this.sessionAggeStatCount
    newMyAcc

  }

  override def reset(): Unit = {

    sessionAggeStatCount = Constants.SESSION_COUNT + "=0|" + Constants.TIME_PERIOD_1s_3s + "=0|" + Constants.TIME_PERIOD_4s_6s + "=0|" + Constants.TIME_PERIOD_7s_9s + "=0|" + Constants.TIME_PERIOD_10s_30s + "=0|" + Constants.TIME_PERIOD_30s_60s + "=0|" + Constants.TIME_PERIOD_1m_3m + "=0|" + Constants.TIME_PERIOD_3m_10m + "=0|" + Constants.TIME_PERIOD_10m_30m + "=0|" + Constants.TIME_PERIOD_30m + "=0|" + Constants.STEP_PERIOD_1_3 + "=0|" + Constants.STEP_PERIOD_4_6 + "=0|" + Constants.STEP_PERIOD_7_9 + "=0|" + Constants.STEP_PERIOD_10_30 + "=0|" + Constants.STEP_PERIOD_30_60 + "=0|" + Constants.STEP_PERIOD_60 + "=0"

  }

  override def add(v: String): Unit = {


    val v1 = sessionAggeStatCount
    val v2 = v
    //    log.warn("v1 : " + v1 + " v2 : " + v2)
    if (StringUtils.isNotEmpty(v1) && StringUtils.isNotEmpty(v2)) {
      var newResult = ""
      // 从v1中，提取v2对应的值，并累加
      val oldValue = StringUtils.getFieldFromConcatString(v1, "\\|", v2)
      if (oldValue != null) {
        val newValue = oldValue.toInt + 1
        newResult = StringUtils.setFieldInConcatString(v1, "\\|", v2, String.valueOf(newValue))
      }
      sessionAggeStatCount = newResult
    }

  }

  override def merge(other: AccumulatorV2[String, String]): Unit =   {

    other match {
      case o : SessionAggrStatAccumulator => {

        sessionAggeStatCount = other.value



      }
      case _ => throw new UnsupportedOperationException(
        s"Cannot merge ${this.getClass.getName} with ${other.getClass.getName}")

    }



  }

  override def value: String = sessionAggeStatCount
}
